/*
 *  Queen Mary University of London MOM System
 *  
 *  Copyright Queen Mary University of London
 *  
 *  Authors:
 *  Bob Chew (bob.chew@elec.qmul.ac.uk)
 *  Beatriz Viñal Murciano (beavimu@gmail.com)
 *  John Bigham (john.bigham@elec.qmul.ac.uk)
 *  
 *  
 *  Portions of this software uses Apache Qpid
 *    http://qpid.apache.org
 *  
 *  Apache Qpid license:
 * --------------------------------------------------------------
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 * -------------------------------------------------------------
 * 
 */
package brokerBottleneckDetector;

import java.io.FileInputStream;
import java.io.IOException;
import java.io.StringReader;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

import javax.jms.JMSException;

import momControlClient.MOMControlClient;

import org.apache.qpid.client.TopicPublisherAdapter;

/**
 * Get the brokers which should be working and create threads to check if each
 * one of them is working well. Send a message to the controller when the
 * working state (=health) of a broker changes.
 */
public class Detector extends MOMControlClient {
	private boolean __DEBUG = false;

	/** Machine in which the detector is running. */
	private final String hostIP;

	/**
	 * List of machines in which a broker is running and their state. State is
	 * an int from WORST_STATE (worst) to BEST_STATE (best)
	 */
	private HashMap<String, Integer> brokersState = new HashMap<String, Integer>();
	/** Worst state of a broker. */
	private static final int WORST_STATE = 0;

	/** Machine in which the controller is running. */
	private final String controllerIP;
	/** Topic used to communicate with the controller. */
	private static final String TOPIC_CONTROLLER = "brokerBottleneck.controller";
	/** Publisher which will send messages to the controller. */
	private TopicPublisherAdapter publisherToController;

	/** JMX port of Brokers */
	private static final String TOP_FILE_NAME = "nwTopology.xml";
	private HashMap<String, Integer> brokersJMX = new HashMap<String, Integer>();

	/**
	 * Constructor.
	 * 
	 * @param hostIP
	 *            IP address of the machine the detector is running on
	 * @param controllerIP
	 *            IP address of the controller
	 * @param brokerAddresses
	 *            list of machines in which a broker is running and the port in
	 *            which they are listening
	 */
	public Detector(final String hostIP, final String controllerIP,
			final HashMap<String, Integer> brokerAddresses) {

		super(); // assign ID

		this.hostIP = hostIP;
		this.controllerIP = controllerIP;

		// start a connection to send messages to the controller
		try {
			// create and start a connection and a session in it
			openConnAndSess(this.controllerIP);
			// create publisher
			publisherToController = createPublisher(TOPIC_CONTROLLER);
		} catch (Exception e) {
			e.printStackTrace();
		}

		try {
			FileInputStream fis = new FileInputStream(TOP_FILE_NAME);
			Properties jmxProps = new Properties();
			jmxProps.loadFromXML(fis);
			int numBrokers = Integer.parseInt(jmxProps
					.getProperty("numBrokers"));
			for (int i = 1; i <= numBrokers; i++) {
				String ip = "broker" + i + "IP";
				String jmx = "broker" + i + "jmx";
				String brokerIP = jmxProps.getProperty(ip);
				int brokerJMX = Integer.parseInt(jmxProps.getProperty(jmx));
				brokersJMX.put(brokerIP, brokerJMX);
			}
			System.out.println("Brokers JMX Ports: " + brokersJMX);
		} catch (Exception e) {
			e.printStackTrace();
		}

		Set<Map.Entry<String, Integer>> brkAddrSet = brokersJMX.entrySet();

		// for each broker
		// set the state AND create a new DetectorThread
		for (Map.Entry<String, Integer> entry : brkAddrSet) {
			// put its initial state (off) in brokersState
			// brokersState.put(entry.getKey(), WORST_STATE);
			// create a thread
			new DetectorThread(this, entry.getKey(), entry.getValue());
		}

		System.out.println("Detector in " + this.hostIP + " ready");

	}

	// receives a set of perf stats from the broker (via onMessage)
	// run whatever detection code needed to classify whether the Broker
	// is in a bottleneck condition
	// 
	// now just run a simple threshold test to trigger the OM message
	protected void testBottleneck(String broker, String t) {
		String[] bAddr = broker.split(":");
		ArrayList<String[]> parsedLog;

		boolean tripped = false;

		try {
			CSVReader cr = new CSVReader(new StringReader(t));
			parsedLog = cr.readAll();
//			if (__DEBUG)
			System.out.println(broker + " ");	
			System.out.println(Arrays.deepToString(parsedLog.toArray())
						.replaceAll("],", "],\n"));

			// ### stub! ###
			// test for cpu exceeding threshold
			// take the average value of all reads, otherwise the spikes will trip the threshold
			double sum = 0.0;
			for (String[] obs : parsedLog) {
				sum += Double.parseDouble(obs[1]);
			}
			double avg = sum / parsedLog.size();
			System.out.println("avg load: " + avg);
			System.out.println("------------------------------------------------------------------------\n");
			
			if (avg >= 20.0 && !tripped) { // ### THRESHOLD
				tripped = true;
				System.out.println("Tripped!");
			}

		} catch (IOException e) {
			System.err.println("Error in CSV parsing.");
			System.err.println(e);
		}

		if (__DEBUG)
			System.out.println("Broker " + bAddr[0] + " is bottlenecked.");

		if (tripped) {
			// report status to the OverlayManager
			updateController(bAddr[0]); // IP only
		}

	}

	public void updateController(String broker) {
		try {
			String text = "bottleneck:" + broker; // eg for broker failure
			if (__DEBUG)
				System.out.println(text);
			// send the message to the super controller
			sendMessage(publisherToController, text);
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}

}
